-module(advisory_plugin).

-include("advisory_plugin.hrl").
-include_lib("emqx/include/emqx.hrl").

-export([register_metrics/0
  , load/0
  , unload/0
]).

-export([on_client_connected/4
  , on_client_disconnected/4
]).

-export([on_client_subscribe/4
  , on_client_unsubscribe/4
]).

-export([on_session_subscribed/4
  , on_session_unsubscribed/4
]).

-export([on_message_publish/2
  , on_message_delivered/3
  , on_message_acked/3
]).

-define(LOG(Level, Format, Args), emqx_logger:Level("advisory_plugin: " ++ Format, Args)).

register_metrics() ->
  [emqx_metrics:new(MetricName) || MetricName <- ['advisory_plugin.client_connected',
    'advisory_plugin.client_disconnected',
    'advisory_plugin.client_subscribe',
    'advisory_plugin.client_unsubscribe',
    'advisory_plugin.session_subscribed',
    'advisory_plugin.session_unsubscribed',
    'advisory_plugin.message_publish',
    'advisory_plugin.message_delivered',
    'advisory_plugin.message_acked']].

load() ->
  io:format("Advisory plugin load...~n"),

  lists:foreach(
    fun({Hook, Fun, Filter}) ->
      load_(Hook, binary_to_atom(Fun, utf8), {Filter})
    end, parse_rule(application:get_env(?APP, hooks, []))).

unload() ->
  io:format("Advisory plugin unload...~n"),
  lists:foreach(
    fun({Hook, Fun, _Filter}) ->
      unload_(Hook, binary_to_atom(Fun, utf8))
    end, parse_rule(application:get_env(?APP, hooks, []))).


%%--------------------------------------------------------------------
%%
%% Client:{
%% anonymous => true,
%% auth_result => success,
%% clientid => <<"1">>,
%% is_bridge => false,
%% is_superuser => false,
%% mountpoint => undefined,
%% peercert => nossl,
%% peerhost => {127,0,0,1},
%% protocol => mqtt,
%% username => <<"userfx1">>,
%% zone => external
%% }
%% See 'Application Message' in MQTT Version 5.0
%%-record(message, {
%%  %% Global unique message ID
%%  id :: binary(),
%%  %% Message QoS
%%  qos = 0,
%%  %% Message from
%%  from :: atom() | binary(),
%%  %% Message flags
%%  flags :: #{atom() => boolean()},
%%  %% Message headers, or MQTT 5.0 Properties
%%  headers :: map(),
%%  %% Topic that the message is published to
%%  topic :: binary(),
%%  %% Message Payload
%%  payload :: binary(),
%%  %% Timestamp
%%  timestamp :: erlang:timestamp()
%%}).
%%--------------------------------------------------------------------
on_client_connected(ClientInfo = #{clientid := ClientId, username := Username, peerhost :=Host}, 0, _ConnInfo, _Env) ->
  emqx_metrics:inc('advisory_plugin.client_connected'),
  Params = [
    {action, connected},
    {clientid, ClientId},
    {username, Username},
    {ip, iolist_to_binary(ntoa(Host))}
  ],
  ?LOG(debug, "on_client_connected ~p~n", [ClientInfo]),
  send_http_post("connected", Params),
  ok;

on_client_connected(#{}, _ConnAck, _ConnInfo, _Env) ->
  ok.


%%--------------------------------------------------------------------
%% Client disconnected
%%--------------------------------------------------------------------
on_client_disconnected(#{}, auth_failure, _ConnInfo, _Env) ->
  ok;


on_client_disconnected(Client, {shutdown, Reason}, ConnInfo, Env) when is_atom(Reason) ->
  on_client_disconnected(Client, Reason, ConnInfo, Env);


on_client_disconnected(ClientInfo = #{clientid := ClientId, username := Username, peerhost :=Host}, Reason, _ConnInfo, _Env)
  when is_atom(Reason) ->
  emqx_metrics:inc('advisory_plugin.client_disconnected'),
  Params = [
    {action, disconnected},
    {clientid, ClientId},
    {username, Username},
    {reason, Reason},
    {ip, iolist_to_binary(ntoa(Host))}
  ],
  ?LOG(debug, "on_client_disconnected ~p~n", [ClientInfo]),

  send_http_post("disconnected", Params),
  ok;
on_client_disconnected(_, Reason, _ConnInfo, _Env) ->
  ?LOG(error, "Client disconnected, cannot encode reason: ~p", [Reason]),
  ok.
%%--------------------------------------------------------------------
%% Client subscribe
%%--------------------------------------------------------------------
on_client_subscribe(#{clientid := ClientId, username := Username}, Properties, RawTopicFilters, {Filter}) ->
  lists:foreach(fun({Topic, Opts}) ->
    with_filter(
      fun() ->
        emqx_metrics:inc('advisory_plugin.client_subscribe'),
        %% Code Start

        %% Here is the code

        %% End
        ok
      end, Topic, Filter)
                end, RawTopicFilters).

%%--------------------------------------------------------------------
%% Client unsubscribe
%%--------------------------------------------------------------------
on_client_unsubscribe(#{clientid := _ClientId, username := _Username}, _Properties, RawTopicFilters, {Filter}) ->
  lists:foreach(fun({Topic, _Opts}) ->
    with_filter(
      fun() ->
        emqx_metrics:inc('advisory_plugin.client_unsubscribe'),
        %% Code Start

        %% Here is the code

        %% End
        ok
      end, Topic, Filter)
                end, RawTopicFilters).

%%--------------------------------------------------------------------
%% Session subscribed
%%--------------------------------------------------------------------
on_session_subscribed(#{clientid := ClientId}, Topic, SubOpts, {Filter}) ->
  with_filter(
    fun() ->
      emqx_metrics:inc('advisory_plugin.session_subscribed'),
      %% Code Start

      %% Here is the code

      %% End
      ok
    end, Topic, Filter).

%%--------------------------------------------------------------------
%% Session unsubscribed
%%--------------------------------------------------------------------
on_session_unsubscribed(#{clientid := _ClientId}, Topic, _Opts, {Filter}) ->
  with_filter(
    fun() ->
      emqx_metrics:inc('advisory_plugin.session_unsubscribed'),
      %% Code Start

      %% Here is the code

      %% End
      ok
    end, Topic, Filter).

%%--------------------------------------------------------------------
%% Message publish
%% {message,
%% <<0,5,152,74,166,126,66,154,244,67,0,0,7,17,0,2>>,
%% 1,
%% <<"1">>,
%% #{dup => false,retain => false},
%% #{allow_publish => true,
%% peerhost => {127,0,0,1},
%% proto_ver => 4,username => <<"userfx1">>},
%% <<"home/garden/fountain">>,
%% <<"111">>,
%% {1574,821271,847580}}

%%--------------------------------------------------------------------
for(Max, Max, F) ->
  [F(Max)];
for(I, Max, F) ->
  [F(I) | for(I + 1, Max, F)].

gen_id(L) when is_list(L) ->
      lists:flatten([erlang:integer_to_list(I) || I <- L]).
    
on_message_publish(Message = #message{topic = <<"$SYS/", _/binary>>}, _Env) ->
  {ok, Message};


on_message_publish(Message = #message{topic = Topic, flags = #{retain := Retain}}, {Filter}) ->
  with_filter(
    fun() ->
      emqx_metrics:inc('advisory_plugin.message_publish'),
      {ClientId, Username} = format_from(Message),
      {message, BinaryMsgId, Qos, ClientId, _Flags, _Headers, Topic, Payload, TimeStrap} = Message,

      Params = [
        {action, publish},
        {messageid, list_to_binary(gen_id(binary_to_list(BinaryMsgId))) },
        {clientid, ClientId},
        {username, Username},
        {topic, Topic},
        {qos, Qos},
        {retain, Retain},
        {payload, encode_payload(Payload)},
        {timestamp, emqx_time:now_secs(TimeStrap)}],
      %% Code Start
      ?LOG(debug, "on_message_publish ClientId:~p Topic ~p Payload:~p ~n", [ClientId, Topic, Payload]),
      send_http_post("publish", Params),
      %% End
      {ok, Message}
    end, Message, Topic, Filter).

%%--------------------------------------------------------------------
%% Message deliver
%%--------------------------------------------------------------------
on_message_delivered(#{clientid := _ClientId, username := _Username},
    _Message = #message{topic = Topic, payload = _Payload},
    {Filter}) ->
  with_filter(
    fun() ->
      emqx_metrics:inc('advisory_plugin.message_delivered'),
      %% Code Start

      %% End
      ok
    end, Topic, Filter).

%%--------------------------------------------------------------------
%% Message acked
%%--------------------------------------------------------------------
on_message_acked(#{clientid := ClientId}, Message = #message{topic = Topic}, {Filter}) ->
  with_filter(
    fun() ->
      emqx_metrics:inc('advisory_plugin.message_acked'),
      %% Code Start

      %% Here is the code

      %% End
      ok
    end, Topic, Filter).

%%--------------------------------------------------------------------
%% Internal functions
%%--------------------------------------------------------------------
parse_rule(Rules) ->
  parse_rule(Rules, []).
parse_rule([], Acc) ->
  lists:reverse(Acc);
parse_rule([{Rule, Conf} | Rules], Acc) ->
  Params = jsx:decode(iolist_to_binary(Conf)),
  Action = proplists:get_value(<<"action">>, Params),
  Filter = proplists:get_value(<<"topic">>, Params),
  parse_rule(Rules, [{list_to_atom(Rule), Action, Filter} | Acc]).

with_filter(Fun, _, undefined) ->
  Fun(), ok;
with_filter(Fun, Topic, Filter) ->
  case emqx_topic:match(Topic, Filter) of
    true -> Fun(), ok;
    false -> ok
  end.

with_filter(Fun, _, _, undefined) ->
  Fun();
with_filter(Fun, Msg, Topic, Filter) ->
  case emqx_topic:match(Topic, Filter) of
    true -> Fun();
    false -> {ok, Msg}
  end.

load_(Hook, Fun, Params) ->
  case Hook of
    'client.connected' -> emqx:hook(Hook, fun ?MODULE:Fun/4, [Params]);
    'client.disconnected' -> emqx:hook(Hook, fun ?MODULE:Fun/4, [Params]);
    'client.subscribe' -> emqx:hook(Hook, fun ?MODULE:Fun/4, [Params]);
    'client.unsubscribe' -> emqx:hook(Hook, fun ?MODULE:Fun/4, [Params]);
    'session.subscribed' -> emqx:hook(Hook, fun ?MODULE:Fun/4, [Params]);
    'session.unsubscribed' -> emqx:hook(Hook, fun ?MODULE:Fun/4, [Params]);
    'message.publish' -> emqx:hook(Hook, fun ?MODULE:Fun/2, [Params]);
    'message.acked' -> emqx:hook(Hook, fun ?MODULE:Fun/3, [Params]);
    'message.delivered' -> emqx:hook(Hook, fun ?MODULE:Fun/3, [Params])
  end.

unload_(Hook, Fun) ->
  case Hook of
    'client.connected' -> emqx:unhook(Hook, fun ?MODULE:Fun/4);
    'client.disconnected' -> emqx:unhook(Hook, fun ?MODULE:Fun/4);
    'client.subscribe' -> emqx:unhook(Hook, fun ?MODULE:Fun/4);
    'client.unsubscribe' -> emqx:unhook(Hook, fun ?MODULE:Fun/4);
    'session.subscribed' -> emqx:unhook(Hook, fun ?MODULE:Fun/4);
    'session.unsubscribed' -> emqx:unhook(Hook, fun ?MODULE:Fun/4);
    'message.publish' -> emqx:unhook(Hook, fun ?MODULE:Fun/2);
    'message.acked' -> emqx:unhook(Hook, fun ?MODULE:Fun/3);
    'message.delivered' -> emqx:unhook(Hook, fun ?MODULE:Fun/3)
  end.

%% IP corvert
ntoa({0, 0, 0, 0, 0, 16#ffff, AB, CD}) ->
  inet_parse:ntoa({AB bsr 8, AB rem 256, CD bsr 8, CD rem 256});
ntoa(IP) ->
  inet_parse:ntoa(IP).

%% Multiple advisory,Current version not support,expect next version
%%advisory(Params) ->
%%  Json = jsx:encode(Params),
%%  Urls = application:get_env(?APP, server, "http://127.0.0.1"),
%%  F = fun(U) ->
%%    {[_N, _K, _S], Url} = U,
%%    case request(post, {Url, [], "application/json", Json}, [{timeout, 5000}], [], 0) of
%%      {ok, _} ->
%%        ?LOG(debug, "Post to:[~p], Params:~s success~n", [Url, Json]),
%%        ok;
%%      {error, Reason} ->
%%        ?LOG(error, "Post to:[~p] failed,Reason:~p ~n", [Url, Reason]),
%%        error
%%    end
%%      end,
%%  lists:foreach(F, Urls),
%%  ok.

%% single post
%% Url+Api
send_http_post(Api, Params) ->
  Json = jsx:encode(Params),
  Url = application:get_env(?APP, destination, "http://127.0.0.1"),
  case request(post, {string:concat(Url, Api), [], "application/json", Json}, [{timeout, 5000}], [], 0) of
    {ok, _} ->
      ?LOG(debug, "Post to:[~p], Params:~s success~n", [Url, Json]),
      ok;
    {error, Reason} ->
      ?LOG(error, "HTTP request error: ~p", [Reason]),
      error
  end.

request(Method, Req, HTTPOpts, Opts, Times) ->
  %% Resend request, when TCP closed by remotely
  case httpc:request(Method, Req, HTTPOpts, Opts) of
    {error, socket_closed_remotely} when Times < 3 ->
      timer:sleep(trunc(math:pow(10, Times))),
      request(Method, Req, HTTPOpts, Opts, Times + 1);
    Other -> Other
  end.
format_from(#message{from = ClientId, headers = #{username := Username}}) ->
  {a2b(ClientId), a2b(Username)};
format_from(#message{from = ClientId, headers = _HeadersNoUsername}) ->
  {a2b(ClientId), <<"undefined">>}.
a2b(A) when is_atom(A) -> erlang:atom_to_binary(A, utf8);
a2b(A) -> A.
encode_payload(Payload) ->
  encode_payload(Payload, application:get_env(?APP, encode_payload, undefined)).

encode_payload(Payload, base62) -> emqx_base62:encode(Payload);
encode_payload(Payload, base64) -> base64:encode(Payload);
encode_payload(Payload, _) -> Payload.
